# @Time   : 2022-03-03 17:00:00
# @Author : Gavin Jiang
# @Owner  : YSRD (Insights Value)

import logging

from flask import current_app
from sqlalchemy import text
from sqlalchemy.orm import sessionmaker

import pandas as pd

from framework.decorator.synchronized import synchronized



class SqlDb:

    @synchronized
    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, '_instance'):
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        self._db = current_app.config['_SQLALCHEMY_INSTANCE']
        self._logger = logging.getLogger(__name__)

    def get_engine(self,bind=None):
        '''
        bind设为None,返回的是默认的数据库 app.config['SQLALCHEMY_DATABASE_URI'],否则为app.config['SQLALCHEMY_BINDS']里,
        '''
        return self._db.get_engine(app=current_app,bind=bind)

    def __good_sql(self, sql):
        if isinstance(sql, str):
            return text(sql)
        return sql

    def __build_in_condition(self, param_prefix, values):
        param_names = []
        params = {}

        index = 0
        for value in values:
            param_name = f"{param_prefix}_{index}"
            param_names.append(f":{param_name}")
            params[param_name] = value

            index = index + 1

        return ','.join(param_names), params

    def __execute(self, connection, sql, params):
        if params and isinstance(params, dict):
            for key, value in params.copy().items():
                if isinstance(sql, str) and (isinstance(value, list) or isinstance(value, tuple)):
                    stmt, stmt_params = self.__build_in_condition(key, value)
                    sql = sql.replace(f":{key}", f"({stmt})")
                    params.update(stmt_params)

        return connection.execute(self.__good_sql(sql), params)

    def list(self, sql, params=None, bind=None):
        with self.get_engine(bind).begin() as connection:
            result = self.__execute(connection, sql, params)
            return [x._asdict() for x in result.all()]

    def pan_list(self, sql, params=None, bind=None):
        with self.get_engine(bind).begin() as connection:
            result = self.__execute(connection, sql, params)
            # 以列表形式返回所有记录的字典表示，每个键都是列名
            return [ row.items()[0][1] for row in result.all()]

    def value(self, sql, params=None, bind=None):
        with self.get_engine(bind).begin() as connection:
            result = self.__execute(connection, sql, params)
            return result.scalar()

    def orm_session(self):
        my_engine = self.get_engine()
        Session = sessionmaker(bind=my_engine)
        db_session = Session()
        return db_session

    def first(self, sql, params=None, bind=None):
        with self.get_engine(bind).begin() as connection:
            result = self.__execute(connection, sql, params)
            data = result.first()
            if data is None:
                return None
            return data._asdict()

    def __process_value(self, value):
        if value is None:
            return 'null'

        if isinstance(value, str):
            value = value.replace("'", "''")
            return f"'{value}'"

        return value

    def __build_sql_for_insert(self, table_name, col_names, col_values, replace):
        col_str = ','.join([f"`{x}`" for x in col_names])

        if not (col_names and col_values):
            self._logger.debug(f"col_names({col_names})和col_values({col_values})不能为空，请检查！")
            return False

        sql = f"{'replace' if replace else 'insert'} into {table_name} ({col_str}) values"

        if isinstance(col_values[0], tuple):
            sql = sql + ','.join([f"({','.join(self.__process_value(x) for x in item)})" for item in col_values])
        else:
            sql = sql + f"({','.join(self.__process_value(x) for x in col_values)})"

        return sql

    def insert(self, table_name, data, method='append', bind=None):
        """
        从原有的拼接sql修改为用pandas，可以方式sql注入
        Args:
            table_name:表名
            data:需要插入的数据，list类型
            method: append,replace,fail
        Returns:
            如果插入成功返回True，否则返回错误信息
        """
        if not isinstance(data, list):
            self._logger.warning(f"data应为list时，略过此元素")

        try:
            df = pd.DataFrame(data)

            print(df)
            with self.get_engine(bind).begin() as connection:
                df.to_sql(name=table_name, con=self.get_engine(bind), if_exists=method, index=False)
        except Exception as e:
            return e

        return True

    def update(self, table_name: str, data: dict, where: str, params=None, bind=None,type=None,sql=None):
        """
        更新数据
        Args:
            table_name:表名
            data:需要更新的数据，字典类型
            where:更新条件，严禁拼接参数构造条件字符串，使用占位符或者占位名
            params:参数

        Returns:
            如果更新成功，返回True，否则返回False
        """
        if type==0:
            with self.get_engine(bind).begin() as connection:
                connection.execute(sql, **params)
            return True
        else:
            if not where:
                self._logger.error("where必须非空，更新数据要传递条件")
                return False

            sql = f"update {table_name} set "
            sql = sql + ",".join([f"`{key}`={self.__process_value(value)}" for key,value in data.items()])
            sql = f"{sql} where {where}"

            self._logger.debug(f"update sql: {sql}")

            with self.get_engine(bind).begin() as connection:
                if params is None:
                    params = {}
                result = connection.execute(text(sql), **params)

                return result.rowcount > 0

    def delete(self, table_name: str, where: str, params=None, bind=None):
        if not where:
            self._logger.error("where必须非空，删除数据要传递条件")

        sql = f"delete from {table_name} where {where}"

        self._logger.debug(f"delete sql: {sql}")

        with self.get_engine(bind).begin() as connection:
            if params is None:
                params = {}
            result = connection.execute(text(sql), **params)

            return result.rowcount > 0

    def query(self, sql, params=None, bind=None):
        with self.get_engine(bind).begin() as connection:
            self.__execute(connection, sql, params)

    def insert_else(self, table_name, data, bind=None):
        if data:
            columns = ','.join([f"`{key}`" for key in data.keys()])
            values = ','.join([f"{self.__process_value(value)}" for value in data.values()])
            sql = f"insert into {table_name}({columns}) values({values})"

            self._logger.debug(f"insert sql: {sql}")

            with self.get_engine(bind).begin() as connection:
                result = connection.execute(text(sql))
                return result.lastrowid

    def bulk_insert(self, table_name, data, bind=None):
        values = ', '.join([f"({', '.join(str(self.__process_value(value)) for value in row.values())})" for row in data])
        columns = ', '.join(data[0].keys())
        sql = f"REPLACE INTO {table_name}({columns}) VALUES {values}"

        self._logger.debug(f"bulk insert sql: {sql}")
        with self.get_engine(bind).begin() as connection:
            result = connection.execute(text(sql))
            return result.rowcount > 0
